package com.example.demo.rx.cat;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.io.InputStream;

public class DownloadService {

    public static Observable<String> download(String urlString) {
        return Observable.create(emitter -> {
            try {
                HttpClient client = HttpClient.newHttpClient();
                HttpRequest request = HttpRequest.newBuilder()
                        .uri(URI.create(urlString))
                        .GET()
                        .build();
                HttpResponse<InputStream> response = client.send(
                        request,
                        HttpResponse.BodyHandlers.ofInputStream()
                );
                InputStream inputStream = response.body();
                byte[] buffer = new byte[50];
                int bytesRead;

                while ((bytesRead = inputStream.read(buffer)) != -1) {
                    if (!emitter.isDisposed()) {
                        String chunk = new String(buffer, 0, bytesRead);
                        emitter.onNext(chunk);
                    } else {
                        break;
                    }
                }

                if (!emitter.isDisposed()) {
                    emitter.onComplete();
                }
            } catch (Exception e) {
                if (!emitter.isDisposed()) {
                    emitter.onError(e);
                }
            }
        });
    }

    public static void main(String[] args) {
        DownloadService.download("http://httpbin.org/get")
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        System.out.println("Subscription started.");
                    }

                    @Override
                    public void onNext(@NonNull String s) {
                        System.out.print(s);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        System.out.println(e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Download completed.");
                    }
                });
    }
}